package world.snowcrystal.job.cycle;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import world.snowcrystal.esdao.PostEsDao;
import world.snowcrystal.model.dto.post.PostEsDTO;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Component
public class CanalSyncToEs implements ApplicationContextAware {

    @Resource
    private PostEsDao postEsDao;

    @Resource
    private CanalConnector connector;

    private final AtomicBoolean running = new AtomicBoolean(false);
    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(@NotNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    record Tuple(CanalEntry.EventType eventType, List<CanalEntry.RowData> rowData) {
    }

    @Scheduled(fixedRate = 60 * 1000)
    public void run() throws Exception {
        if (!running.compareAndSet(false,true)) {
            return;
        }
        if (!connector.checkValid()) {
            connector = applicationContext.getBean(CanalConnector.class);
        }
        int batchSize = 1000;
//        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
//            int totalEmptyCount = 120;
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                final List<CanalEntry.Entry> entries = message.getEntries();
                long batchId = message.getId();
                int size = entries.size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000L);
                        Thread.onSpinWait();
                    } catch (InterruptedException ignored) {
                    }
                } else {
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    sync(entries);
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } catch (Exception e) {
            log.error("an error occurred . Canal2Es ended {}",e.getClass().getName());
            connector.disconnect();
            e.printStackTrace();
        } finally {
            running.set(false);
        }
    }

    private void sync(List<CanalEntry.Entry> entrys) {
        entrys.stream()
                .filter(e -> e.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN
                        && e.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND)
                .map(e -> {
                    final CanalEntry.RowChange rowChange;
                    try {
                        rowChange = CanalEntry.RowChange.parseFrom(e.getStoreValue());
                    } catch (InvalidProtocolBufferException ex) {
                        throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + e,
                                ex);
                    }
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    log.debug(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                            e.getHeader().getLogfileName(), e.getHeader().getLogfileOffset(),
                            e.getHeader().getSchemaName(), e.getHeader().getTableName(),
                            eventType));
                    return new Tuple(eventType, rowChange.getRowDatasList());
                })
                .forEach(tuple -> {
                    for (CanalEntry.RowData rowData : tuple.rowData()) {
                        switch (tuple.eventType()) {
                            case DELETE -> delete(rowData);
                            case INSERT, UPDATE -> save(rowData);
                            default -> {
                                log.debug("-------&gt; before");
                                printColumn(rowData.getBeforeColumnsList());
                                log.debug("-------&gt; after");
                                printColumn(rowData.getAfterColumnsList());
                            }
                        }
                    }
                });


        /////////////////////
//        for (CanalEntry.Entry entry : entrys) {
//            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
//                continue;
//            }
//            CanalEntry.RowChange rowChage = null;
//            try {
//                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
//            } catch (Exception e) {
//                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
//                        e);
//            }
//
//            CanalEntry.EventType eventType = rowChage.getEventType();
//            log.info(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
//                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
//                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
//                    eventType));
//
//            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
//                if (eventType == CanalEntry.EventType.DELETE) {
//                    printColumn(rowData.getBeforeColumnsList());
//                } else if (eventType == CanalEntry.EventType.INSERT) {
//                    printColumn(rowData.getAfterColumnsList());
//                } else {
//                    System.out.println("-------&gt; before");
//                    printColumn(rowData.getBeforeColumnsList());
//                    System.out.println("-------&gt; after");
//                    printColumn(rowData.getAfterColumnsList());
//                }
//            }
//
//
//
//        }
    }

    private static void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            log.debug("{} : {}    update={}", column.getName(), column.getValue(), column.getUpdated());
        }
    }

    private static PostEsDTO parse(List<CanalEntry.Column> columns) {
        PostEsDTO dto = new PostEsDTO();
        columns.forEach(column -> setDto(column, dto));
        return dto;
    }

    private void delete(CanalEntry.RowData rowData) {
        final List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
        printColumn(beforeColumnsList);
        beforeColumnsList.stream()
                .filter(column ->
                        "id".equals(column.getName())
                )
                .map(column -> Long.parseLong(column.getValue()))
                .forEach(id -> {
                    postEsDao.deleteById(id);
                    log.info("delete data id={}", id);
                });
    }

    private void save(CanalEntry.RowData rowData) {
        final List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        printColumn(afterColumnsList);
        final PostEsDTO dto = parse(afterColumnsList);
        postEsDao.save(dto);
        log.info("insert or update data id={}", dto.getId());
    }

    @SneakyThrows
    private static void setDto(CanalEntry.Column column, PostEsDTO dto)  {
        final String name = column.getName();
        final String value = column.getValue();
        final Class<PostEsDTO> dtoClass = PostEsDTO.class;
        Method declaredMethod ;
        String setter = "set"
                + name.replaceFirst(String.valueOf(name.charAt(0)),
                                                    String.valueOf(Character.toUpperCase(name.charAt(0))));
        if(!hasSuchField(dtoClass, name)){
            return;
        }
        try{
            declaredMethod = dtoClass.getDeclaredMethod(setter,String.class);
        }catch (NoSuchMethodException e){
            // setter 策略模式
            declaredMethod = dtoClass.getDeclaredMethod(setter+"Strategy",String.class);
        }
        declaredMethod.invoke(dto,value);
    }
    private static boolean hasSuchField(Class<?> clazz, String fieldName){
        try{
            clazz.getDeclaredField(fieldName);
            return true;
        }catch (NoSuchFieldException e){
            return false;
        }
    }

}
